<!DOCTYPE html>



  


<html class="theme-next mist use-motion" lang="zh-Hans">
<head>
  <meta charset="UTF-8"/>
<meta http-equiv="X-UA-Compatible" content="IE=edge" />
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1"/>
<meta name="theme-color" content="#222">









<meta http-equiv="Cache-Control" content="no-transform" />
<meta http-equiv="Cache-Control" content="no-siteapp" />
















  
  
  <link href="/jexo/lib/fancybox/source/jquery.fancybox.css?v=2.1.5" rel="stylesheet" type="text/css" />







<link href="/jexo/lib/font-awesome/css/font-awesome.min.css?v=4.6.2" rel="stylesheet" type="text/css" />

<link href="/jexo/css/main.css?v=5.1.4" rel="stylesheet" type="text/css" />


  <link rel="apple-touch-icon" sizes="180x180" href="/jexo/images/apple-touch-icon-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="32x32" href="/jexo/images/favicon-32x32-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="16x16" href="/jexo/images/favicon-16x16-next.png?v=5.1.4">


  <link rel="mask-icon" href="/jexo/images/logo.svg?v=5.1.4" color="#222">





  <meta name="keywords" content="Android,RxJava,源码解析," />










<meta name="description" content="Amb操作符功能介绍 参数传入多个Observable,哪个Observable中的OnSubscribe执行的快就执行就执行哪个，而其他Observable都被取消">
<meta name="keywords" content="Android,RxJava,源码解析">
<meta property="og:type" content="article">
<meta property="og:title" content="看看谁最快的操作符-----Amb">
<meta property="og:url" content="http://ja_project.gitee.io/jexo/2018/04/18/看看谁最快的操作符-Amb/index.html">
<meta property="og:site_name" content="Ling Hu Blog">
<meta property="og:description" content="Amb操作符功能介绍 参数传入多个Observable,哪个Observable中的OnSubscribe执行的快就执行就执行哪个，而其他Observable都被取消">
<meta property="og:locale" content="zh-Hans">
<meta property="og:updated_time" content="2018-04-18T13:47:04.000Z">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="看看谁最快的操作符-----Amb">
<meta name="twitter:description" content="Amb操作符功能介绍 参数传入多个Observable,哪个Observable中的OnSubscribe执行的快就执行就执行哪个，而其他Observable都被取消">



<script type="text/javascript" id="hexo.configurations">
  var NexT = window.NexT || {};
  var CONFIG = {
    root: '/jexo/',
    scheme: 'Mist',
    version: '5.1.4',
    sidebar: {"position":"left","display":"post","offset":12,"b2t":false,"scrollpercent":false,"onmobile":false},
    fancybox: true,
    tabs: true,
    motion: {"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"slideDownIn","post_body":"slideDownIn","coll_header":"slideLeftIn","sidebar":"slideUpIn"}},
    duoshuo: {
      userId: '0',
      author: '博主'
    },
    algolia: {
      applicationID: '',
      apiKey: '',
      indexName: '',
      hits: {"per_page":10},
      labels: {"input_placeholder":"Search for Posts","hits_empty":"We didn't find any results for the search: ${query}","hits_stats":"${hits} results found in ${time} ms"}
    }
  };
</script>



  <link rel="canonical" href="http://ja_project.gitee.io/jexo/2018/04/18/看看谁最快的操作符-Amb/"/>





  <title>看看谁最快的操作符-----Amb | Ling Hu Blog</title>
  





  <script type="text/javascript">
    var _hmt = _hmt || [];
    (function() {
      var hm = document.createElement("script");
      hm.src = "https://hm.baidu.com/hm.js?abc8796c232e2b898d516245257842b4";
      var s = document.getElementsByTagName("script")[0];
      s.parentNode.insertBefore(hm, s);
    })();
  </script>




</head>

<body itemscope itemtype="http://schema.org/WebPage" lang="zh-Hans">

  
  
    
  

  <div class="container sidebar-position-left page-post-detail">
    <div class="headband"></div>

    <header id="header" class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-wrapper">
  <div class="site-meta ">
    

    <div class="custom-logo-site-title">
      <a href="/jexo/"  class="brand" rel="start">
        <span class="logo-line-before"><i></i></span>
        <span class="site-title">Ling Hu Blog</span>
        <span class="logo-line-after"><i></i></span>
      </a>
    </div>
      
        <p class="site-subtitle"></p>
      
  </div>

  <div class="site-nav-toggle">
    <button>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
    </button>
  </div>
</div>

<nav class="site-nav">
  

  
    <ul id="menu" class="menu">
      
        
        <li class="menu-item menu-item-home">
          <a href="/jexo/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-home"></i> <br />
            
            首页
          </a>
        </li>
      
        
        <li class="menu-item menu-item-about">
          <a href="/jexo/about/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-user"></i> <br />
            
            关于
          </a>
        </li>
      
        
        <li class="menu-item menu-item-tags">
          <a href="/jexo/tags/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-tags"></i> <br />
            
            标签
          </a>
        </li>
      
        
        <li class="menu-item menu-item-categories">
          <a href="/jexo/categories/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-th"></i> <br />
            
            分类
          </a>
        </li>
      
        
        <li class="menu-item menu-item-archives">
          <a href="/jexo/archives/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-archive"></i> <br />
            
            归档
          </a>
        </li>
      

      
    </ul>
  

  
</nav>



 </div>
    </header>

    <main id="main" class="main">
      <div class="main-inner">
        <div class="content-wrap">
          <div id="content" class="content">
            

  <div id="posts" class="posts-expand">
    

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="http://ja_project.gitee.io/jexo/jexo/2018/04/18/看看谁最快的操作符-Amb/">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="Ling Hu">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/jexo/images/avatar.jpg">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="Ling Hu Blog">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">看看谁最快的操作符-----Amb</h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2018-04-18T21:46:43+08:00">
                2018-04-18
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/jexo/categories/RxJava操作符源码解析/" itemprop="url" rel="index">
                    <span itemprop="name">RxJava操作符源码解析</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
              <span class="post-comments-count">
                <span class="post-meta-divider">|</span>
                <span class="post-meta-item-icon">
                  <i class="fa fa-comment-o"></i>
                </span>
                <a href="/jexo/2018/04/18/看看谁最快的操作符-Amb/#comments" itemprop="discussionUrl">
                  <span class="post-comments-count gitment-comments-count" data-xid="/jexo/2018/04/18/看看谁最快的操作符-Amb/" itemprop="commentsCount"></span>
                </a>
              </span>
            
          

          
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        <h6 id="Amb操作符功能介绍"><a href="#Amb操作符功能介绍" class="headerlink" title="Amb操作符功能介绍"></a>Amb操作符功能介绍</h6><p> 参数传入多个<code>Observable</code>,哪个<code>Observable</code>中的<code>OnSubscribe</code>执行的快就执行就执行哪个，而其他<code>Observable</code>都被取消</p>
<a id="more"></a>
<h6 id="用途"><a href="#用途" class="headerlink" title="用途"></a>用途</h6><p>这个操作符，其实我不常用，但是用途大家可以自己想象下。<br>有多个异步操作，你也不知道哪个会先执行完毕，你只想在某一个操作执行完毕后，执行一次某个方法，那么这个操作符就刚好适用。</p>
<h6 id="来一段小代码"><a href="#来一段小代码" class="headerlink" title="来一段小代码"></a>来一段小代码</h6><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line">  observable = Observable.amb(Observable.create(<span class="keyword">object</span> : Observable.OnSubscribe&lt;String&gt; &#123;</span><br><span class="line">            <span class="keyword">override</span> <span class="function"><span class="keyword">fun</span> <span class="title">call</span><span class="params">(t: <span class="type">Subscriber</span>&lt;<span class="type">in</span> <span class="type">String</span>&gt;)</span></span> &#123;</span><br><span class="line">                t.onNext(<span class="string">"Test1"</span>)</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;), Observable.create(<span class="keyword">object</span> : Observable.OnSubscribe&lt;String&gt; &#123;</span><br><span class="line">            <span class="keyword">override</span> <span class="function"><span class="keyword">fun</span> <span class="title">call</span><span class="params">(t: <span class="type">Subscriber</span>&lt;<span class="type">in</span> <span class="type">String</span>&gt;)</span></span> &#123;</span><br><span class="line">                t.onNext(<span class="string">"Test2"</span>)</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;))</span><br><span class="line">...</span><br><span class="line"> observable?.subscribe(&#123; msg -&gt;</span><br><span class="line">            tvContent.text = tvContent.text.toString() + <span class="string">"\n"</span> + msg</span><br><span class="line">        &#125;)</span><br></pre></td></tr></table></figure>
<p>这里我们创建了2个<code>Observable</code>传入<code>amb</code>操作符。<br>然后<code>subscribe</code>订阅消息。<br>demo很简单，就不过多介绍了，下面直接看源码分析。</p>
<hr>
<h6 id="看看源代码"><a href="#看看源代码" class="headerlink" title="看看源代码"></a>看看源代码</h6><p>Observable<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> static &lt;T&gt; Observable&lt;T&gt; amb(Observable&lt;? extends T&gt; o1, Observable&lt;? extends T&gt; o2) &#123;</span><br><span class="line">       <span class="keyword">return</span> unsafeCreate(OnSubscribeAmb.amb(o1, o2));</span><br><span class="line">   &#125;</span><br><span class="line"></span><br><span class="line"><span class="keyword">public</span> static &lt;T&gt; Observable&lt;T&gt; unsafeCreate(OnSubscribe&lt;T&gt; f) &#123;</span><br><span class="line">       <span class="keyword">return</span> new Observable&lt;T&gt;(RxJavaHooks.onCreate(f));</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure></p>
<p>如果大家看过上一篇文章，会发现其实<code>Observable.create</code>和<code>Observable.unsafeCreate</code><br>是一致的，没有区别。<br>Observable<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line"> <span class="keyword">public</span> static &lt;T&gt; Observable&lt;T&gt; create(OnSubscribe&lt;T&gt; f) &#123;</span><br><span class="line">        <span class="keyword">return</span> new Observable&lt;T&gt;(RxJavaHooks.onCreate(f));</span><br><span class="line">    &#125;</span><br><span class="line">...</span><br><span class="line"><span class="keyword">public</span> static &lt;T&gt; Observable&lt;T&gt; unsafeCreate(OnSubscribe&lt;T&gt; f) &#123;</span><br><span class="line">        <span class="keyword">return</span> new Observable&lt;T&gt;(RxJavaHooks.onCreate(f));</span><br><span class="line">    &#125;</span><br><span class="line">...</span><br></pre></td></tr></table></figure></p>
<p>从这几行代码中，我们可以看出，最重要的其实是这一行<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">OnSubscribeAmb.amb(o1, o2)</span><br></pre></td></tr></table></figure></p>
<p>先不看源码，我们可以大致的了解，这句话的作用就是创建一个 <code>OnSubscribe</code>对象，根据上一篇文章的介绍，其实可以了解到我们最重要的就是看2个东西<code>OnSubscribe</code>的<code>call</code>,还有就是<code>Subscriber</code>对象的<code>onNext</code>,<code>onComplete</code>…等方法</p>
<p>下面我们继续深入<br>OnSubscribeAmb<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line"><span class="keyword">public</span> static &lt;T&gt; OnSubscribe&lt;T&gt; amb(Observable&lt;? extends T&gt; o1, Observable&lt;? extends T&gt; o2) &#123;</span><br><span class="line">        List&lt;Observable&lt;? extends T&gt;&gt; sources = new ArrayList&lt;Observable&lt;? extends T&gt;&gt;();</span><br><span class="line">        sources.add(o1);</span><br><span class="line">        sources.add(o2);</span><br><span class="line">        <span class="keyword">return</span> amb(sources);</span><br><span class="line">    &#125;</span><br><span class="line">...</span><br><span class="line"><span class="keyword">public</span> static &lt;T&gt; OnSubscribe&lt;T&gt; amb(<span class="keyword">final</span> Iterable&lt;? extends Observable&lt;? extends T&gt;&gt; sources) &#123;</span><br><span class="line">        <span class="keyword">return</span> new OnSubscribeAmb&lt;T&gt;(sources);</span><br><span class="line">    &#125;</span><br><span class="line">...</span><br><span class="line"><span class="keyword">public</span> void call(<span class="keyword">final</span> Subscriber&lt;? <span class="keyword">super</span> T&gt; subscriber) &#123;</span><br><span class="line">        <span class="keyword">final</span> Selection&lt;T&gt; selection = new Selection&lt;T&gt;();</span><br><span class="line">        subscriber.add(Subscriptions.create(new Action0() &#123;</span><br><span class="line"></span><br><span class="line">            <span class="meta">@Override</span></span><br><span class="line">            <span class="keyword">public</span> void call() &#123;</span><br><span class="line">                AmbSubscriber&lt;T&gt; c;</span><br><span class="line">                <span class="keyword">if</span> ((c = selection.<span class="keyword">get</span>()) != <span class="literal">null</span>) &#123;</span><br><span class="line">                    c.unsubscribe();</span><br><span class="line">                &#125;</span><br><span class="line">                unsubscribeAmbSubscribers(selection.ambSubscribers);</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">        &#125;));</span><br><span class="line"></span><br><span class="line">        <span class="keyword">for</span> (Observable&lt;? extends T&gt; source : sources) &#123;</span><br><span class="line">            <span class="keyword">if</span> (subscriber.isUnsubscribed()) &#123;</span><br><span class="line">                <span class="keyword">break</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            AmbSubscriber&lt;T&gt; ambSubscriber = new AmbSubscriber&lt;T&gt;(<span class="number">0</span>, subscriber, selection);</span><br><span class="line">            selection.ambSubscribers.add(ambSubscriber);</span><br><span class="line"></span><br><span class="line">            AmbSubscriber&lt;T&gt; c;</span><br><span class="line">            <span class="keyword">if</span> ((c = selection.<span class="keyword">get</span>()) != <span class="literal">null</span>) &#123;</span><br><span class="line">                <span class="comment">// Already chose one, the rest can be skipped and we can clean up</span></span><br><span class="line">                selection.unsubscribeOthers(c);</span><br><span class="line">                <span class="keyword">return</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            source.unsafeSubscribe(ambSubscriber);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">// while subscribing unsubscription may have occurred so we clean up after</span></span><br><span class="line">        <span class="keyword">if</span> (subscriber.isUnsubscribed()) &#123;</span><br><span class="line">            unsubscribeAmbSubscribers(selection.ambSubscribers);</span><br><span class="line">        &#125;</span><br><span class="line">      ...</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>这段代码有点长，最主要的当然是<code>call</code>方法，其中<code>amb</code>重载方法中，主要是new 了就是new了一个<code>OnSubscribe</code>没什么好说的。</p>
<p>最终呢还是会执行<code>call</code>,所以我们来重点分析下<code>call</code>方法。</p>
<blockquote>
<p>这里我又要提出一个个人的阅读源码的方法，如果已经了解了这个方法或者这个类的功能，那么就尝试着想象自己大概会怎么去实现。</p>
</blockquote>
<p>比如这里，如果是我，我会这么去实现，遍历我的List&lt; Observable &gt;，然后一个个调用<code>subscribe</code>去运行，当第一个<code>OnSubscribe</code>调用完<code>call</code>后，把其他<code>Observable</code>都取消。</p>
<p>下面我们继续深入,call方法中有个<code>Selection</code>,我们来看下具体是什么东西</p>
<p>OnSubscribeAmb<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br></pre></td><td class="code"><pre><span class="line">static <span class="keyword">final</span> <span class="class"><span class="keyword">class</span> <span class="title">Selection</span>&lt;<span class="type">T</span>&gt; <span class="title">extends</span> <span class="title">AtomicReference</span>&lt;<span class="type">AmbSubscriber&lt;T</span>&gt;&gt; </span>&#123;</span><br><span class="line">      <span class="keyword">final</span> Collection&lt;AmbSubscriber&lt;T&gt;&gt; ambSubscribers = new ConcurrentLinkedQueue&lt;AmbSubscriber&lt;T&gt;&gt;();</span><br><span class="line"></span><br><span class="line">      <span class="keyword">public</span> void unsubscribeLosers() &#123;</span><br><span class="line">          AmbSubscriber&lt;T&gt; winner = <span class="keyword">get</span>();</span><br><span class="line">          <span class="keyword">if</span> (winner != <span class="literal">null</span>) &#123;</span><br><span class="line">              unsubscribeOthers(winner);</span><br><span class="line">          &#125;</span><br><span class="line">      &#125;</span><br><span class="line"></span><br><span class="line">      <span class="keyword">public</span> void unsubscribeOthers(AmbSubscriber&lt;T&gt; notThis) &#123;</span><br><span class="line">          <span class="keyword">for</span> (AmbSubscriber&lt;T&gt; other : ambSubscribers) &#123;</span><br><span class="line">              <span class="keyword">if</span> (other != notThis) &#123;</span><br><span class="line">                  other.unsubscribe();</span><br><span class="line">              &#125;</span><br><span class="line">          &#125;</span><br><span class="line">          ambSubscribers.clear();</span><br><span class="line">      &#125;</span><br><span class="line"></span><br><span class="line">  &#125;</span><br></pre></td></tr></table></figure></p>
<p>前面介绍了了Amb操作符可能使用到的场景就是有很多的异步操作，那么当然也会有多线程的问题，所以这里使用了<code>Atomic**</code>，这里就不深入去介绍了，有兴趣的自己去百度。</p>
<p>我们可以看下<code>Selection</code>的主要作用</p>
<ol>
<li>有一个变量<code>ambSubscribers</code>，用来存储<code>Subscriber</code>对象，在这具体的实现是<code>AmbSubscriber</code>。</li>
<li>取消订阅其他的<code>Subscriber</code><br>看完了<code>Selection</code>，那么我们再回过去看<code>call</code>方法中的代码。</li>
</ol>
<figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">//第一部分</span></span><br><span class="line"><span class="keyword">final</span> Selection&lt;T&gt; selection = new Selection&lt;T&gt;();</span><br><span class="line"><span class="comment">//第二部分</span></span><br><span class="line"> subscriber.add(Subscriptions.create...));</span><br><span class="line"><span class="comment">//第三部分</span></span><br><span class="line"> <span class="keyword">for</span> (Observable&lt;? extends T&gt; source : sources) &#123;&#125;</span><br><span class="line"><span class="comment">//第四部分</span></span><br><span class="line"><span class="keyword">if</span> (subscriber.isUnsubscribed()) &#123;</span><br><span class="line">            unsubscribeAmbSubscribers(selection.ambSubscribers);</span><br><span class="line">        &#125;</span><br><span class="line"><span class="comment">//第五部分</span></span><br><span class="line">subscriber.setProducer(...)</span><br></pre></td></tr></table></figure>
<p>这样子标明代码后，其实我们只要一个一个去了解它的意思就好了<br>第一部分只是创建了一个<code>Selection</code>，就不过多介绍了，下面我们来看第二部分。</p>
<h6 id="第二部分"><a href="#第二部分" class="headerlink" title="第二部分"></a>第二部分</h6><p>第二部分呢其实也非常的简单，只是往<code>Subscriber</code>中添加了一个<code>Subscription</code></p>
<p>Subscriber<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line"> <span class="keyword">public</span> <span class="keyword">final</span> void add(Subscription s) &#123;</span><br><span class="line">        subscriptions.add(s);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">final</span> void unsubscribe() &#123;</span><br><span class="line">        subscriptions.unsubscribe();</span><br><span class="line">    &#125;</span><br><span class="line">...</span><br></pre></td></tr></table></figure></p>
<p><code>Subscriber</code>中本身保存了一个<code>List&lt;Subscription&gt;</code>，当这个<code>Subscriber</code>取消注册的时候，把<code>List&lt;Subscription&gt;</code>中的所有的<code>Subscription</code>取消。</p>
<p>第三部分</p>
<p>这一部分，其实跟我前面所做的猜想是一致的。所以大家有时间的时候不妨也试一试，也是很锻炼的。<br>我们来详细的看下第三部分的代码。<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">for</span> (Observable&lt;? extends T&gt; source : sources) &#123;</span><br><span class="line">            <span class="keyword">if</span> (subscriber.isUnsubscribed()) &#123;</span><br><span class="line">                <span class="keyword">break</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            AmbSubscriber&lt;T&gt; ambSubscriber = new AmbSubscriber&lt;T&gt;(<span class="number">0</span>, subscriber, selection);</span><br><span class="line">            selection.ambSubscribers.add(ambSubscriber);</span><br><span class="line">            AmbSubscriber&lt;T&gt; c;</span><br><span class="line">            <span class="keyword">if</span> ((c = selection.<span class="keyword">get</span>()) != <span class="literal">null</span>) &#123;</span><br><span class="line">                selection.unsubscribeOthers(c);</span><br><span class="line">                <span class="keyword">return</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            source.unsafeSubscribe(ambSubscriber);</span><br><span class="line">        &#125;</span><br></pre></td></tr></table></figure></p>
<p>源代码中本身有很多注释，为了节省空间，我把注释都去掉了，<br>在这段代码中，最重要的一句是<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">AmbSubscriber&lt;T&gt; ambSubscriber = new AmbSubscriber&lt;T&gt;(<span class="number">0</span>, subscriber, selection);</span><br></pre></td></tr></table></figure></p>
<p>前面我一直在强调，其实源码的阅读，最重要的是看<code>OnSubscribe</code>和<code>Subsciber</code>的子类。源代码中经常会在原来的<code>OnSubscribe</code>和<code>Subsciber</code>给你外面套上一层，你要是没仔细看的话，就不能完全看懂。</p>
<p>所以这里也是，使用了<code>AmbSubscriber</code>给原来的<code>Subsciber</code>包了一层，所以当我们<code>subscribe</code>订阅消息的时候，是先调用了<code>AmbSubscriber</code>中的<code>onNext</code>等方法，然后才会调用了我们自己的<code>Subscriber</code>。</p>
<p>下面一句是<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">if</span> ((c = selection.<span class="keyword">get</span>()) != <span class="literal">null</span>) &#123;</span><br><span class="line">                selection.unsubscribeOthers(c);</span><br><span class="line">                <span class="keyword">return</span>;</span><br><span class="line">            &#125;</span><br></pre></td></tr></table></figure></p>
<p><code>selection.get()</code>出来的是<code>AmbSubscriber</code>对象，如果这个对象不为空，就说明，已经有一个<code>OnSubscibe</code>执行完成了，这其实是我的一个结论，就前面的代码我们还无法分析出这个结论，但是这里我先把结论给出，在后面我们可以再回过来看。会更加明白。</p>
<p>最后一句<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">source.unsafeSubscribe(ambSubscriber);</span><br></pre></td></tr></table></figure></p>
<p>source其实就是我们调用<code>amb</code>操作符的时候传入的<code>Observable</code>。调用用了<code>unsafeSubscribe</code>方法。</p>
<p>Observable<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">final</span> Subscription unsafeSubscribe(Subscriber&lt;? <span class="keyword">super</span> T&gt; subscriber) &#123;</span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            subscriber.onStart();</span><br><span class="line">            RxJavaHooks.onObservableStart(<span class="keyword">this</span>, onSubscribe).call(subscriber);</span><br><span class="line">            <span class="keyword">return</span> RxJavaHooks.onObservableReturn(subscriber);</span><br><span class="line">...</span><br></pre></td></tr></table></figure></p>
<p>这个方法跟前面一篇文章提到的<code>subscribe</code>是很像的，区别就在于没有另外再把参数<code>Subscriber</code>包一层<code>SafeSubscriber</code>。</p>
<p>所以最后也是一样的调用的是<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">observable. onSubscribe.call(subscriber)</span><br></pre></td></tr></table></figure></p>
<p>而这里的<code>subscriber</code>其实是<code>AmbSubscriber</code>,<code>onSubscribe</code>其实就是我们在demo中传入的<code>OnSubscribe</code>对象。下面我们再来看下demo的代码<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line">  observable = Observable.amb(Observable.create(<span class="keyword">object</span> : Observable.OnSubscribe&lt;String&gt; &#123;</span><br><span class="line">            <span class="keyword">override</span> <span class="function"><span class="keyword">fun</span> <span class="title">call</span><span class="params">(t: <span class="type">Subscriber</span>&lt;<span class="type">in</span> <span class="type">String</span>&gt;)</span></span> &#123;</span><br><span class="line">                t.onNext(<span class="string">"Test1"</span>)</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;), Observable.create(<span class="keyword">object</span> : Observable.OnSubscribe&lt;String&gt; &#123;</span><br><span class="line">            <span class="keyword">override</span> <span class="function"><span class="keyword">fun</span> <span class="title">call</span><span class="params">(t: <span class="type">Subscriber</span>&lt;<span class="type">in</span> <span class="type">String</span>&gt;)</span></span> &#123;</span><br><span class="line">                t.onNext(<span class="string">"Test2"</span>)</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;))</span><br><span class="line">...</span><br></pre></td></tr></table></figure></p>
<p>demo中传入的<code>OnSubscribe</code>对象中的<code>call</code>方法写的很简单，直接调用<code>Subscriber</code>中的<code>onNext</code>方法。也就是<code>AmbSubscriber</code>的<code>onNext</code>方法</p>
<p>AmbSubscriber</p>
<figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line"><span class="keyword">public</span> void onNext(T t) &#123;</span><br><span class="line">            <span class="keyword">if</span> (isSelected()) &#123;</span><br><span class="line">                subscriber.onNext(t);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">...</span><br><span class="line"><span class="keyword">private</span> boolean isSelected() &#123;</span><br><span class="line">            <span class="keyword">if</span> (chosen) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="literal">true</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            <span class="keyword">if</span> (selection.<span class="keyword">get</span>() == <span class="keyword">this</span>) &#123;</span><br><span class="line">                <span class="comment">// fast-path</span></span><br><span class="line">                chosen = <span class="literal">true</span>;</span><br><span class="line">                <span class="keyword">return</span> <span class="literal">true</span>;</span><br><span class="line">            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                <span class="keyword">if</span> (selection.compareAndSet(<span class="literal">null</span>, <span class="keyword">this</span>)) &#123;</span><br><span class="line">                    selection.unsubscribeOthers(<span class="keyword">this</span>);</span><br><span class="line">                    chosen = <span class="literal">true</span>;</span><br><span class="line">                    <span class="keyword">return</span> <span class="literal">true</span>;</span><br><span class="line">                &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                    <span class="comment">// we lost so unsubscribe ... and force cleanup again due to possible race conditions</span></span><br><span class="line">                    selection.unsubscribeLosers();</span><br><span class="line">                    <span class="keyword">return</span> <span class="literal">false</span>;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">...</span><br></pre></td></tr></table></figure>
<p><code>AmbSubscriber</code>代码写的非常的简单，我这里简单做个翻译。<br>先判断是否被选中了，如果被选中了，就直接执行<code>Subscriber</code>的<code>onNext</code>，这里的<code>Subscriber</code>也就是我们自己demo中写的<code>onNext</code>的<code>Subscriber</code>(其实是被源码包了一层的<code>ActionSubscriber</code>，这不是重点)。</p>
<p>最重要的代码是这一段<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">if</span> (selection.compareAndSet(<span class="literal">null</span>, <span class="keyword">this</span>)) &#123;</span><br><span class="line">                   selection.unsubscribeOthers(<span class="keyword">this</span>);</span><br><span class="line">                   chosen = <span class="literal">true</span>;</span><br><span class="line">                   <span class="keyword">return</span> <span class="literal">true</span>;</span><br><span class="line">               &#125;</span><br></pre></td></tr></table></figure></p>
<p>selection.get()获取的<code>AmbSubscriber</code>不是this，不是的话，就<code>selection.compareAndSet(null, this)</code>，前面介绍了<code>Selection</code>其实是一个<code>Atomic**</code>的对象，这种对象有个常用的方法就是<code>compareAndSet(null, this)</code>，意思就是如果<code>selection</code>的值null就把<code>this</code>设置为它的值，如果为<code>null</code> 返回<code>true</code>否者返回<code>false</code>。</p>
<p>所以这一段代码其实意思很好理解，第一个<code>AmbSubscriber</code>调用了<code>onNext</code>方法，就会把<code>selection</code>的值设置为this，下一个<code>AmbSubscriber</code>再来执行的话就晚了。并且在第一个<code>AmbSubscriber</code>把<code>selection</code>的值设置为this之后，还会对其他<code>AmbSubscriber</code>取消订阅。</p>
<h6 id="第四部分"><a href="#第四部分" class="headerlink" title="第四部分"></a>第四部分</h6><p>直接可以看出，如果被取消订阅的话，就直接把就<code>List&lt; AmbSubscriber &gt;</code>中所有的<code>Subscriber</code>取消订阅，然后清空。这一步主要是防止用户可能会取消订阅。</p>
<h6 id="第五部分"><a href="#第五部分" class="headerlink" title="第五部分"></a>第五部分</h6><p>这几部分其实第三部分是最重要的，没有4，5部分，我们也对<code>Amb</code>操作符有了深入的理解了。但是第五部分对理解<code>RxJava</code>源码本身，有重要的作用。</p>
<p>OnSubscribeAmb<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"> subscriber.setProducer(new Producer() &#123;</span><br><span class="line">            <span class="meta">@Override</span></span><br><span class="line">            <span class="keyword">public</span> void request(long n) &#123;</span><br><span class="line">            ...</span><br><span class="line">            &#125;</span><br><span class="line">&#125;)</span><br></pre></td></tr></table></figure></p>
<p>Subscriber<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> void setProducer(Producer p) &#123;</span><br><span class="line">       ...</span><br><span class="line">            <span class="keyword">if</span> (toRequest == NOT_SET) &#123;</span><br><span class="line">                producer.request(<span class="built_in">Long</span>.MAX_VALUE);</span><br><span class="line">            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                producer.request(toRequest);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>其他不需要看，我只看到了<code>producer.request</code>，也就是说，其实setProducer就是直接调用了<code>request</code>方法，当然本身没那么简单，但是如果我们是第一次去理解的话，就按最简单的去理解，由浅入深，是很好的一种学习方式。</p>
<p>那么我们回过去再看</p>
<p>OnSubscribeAmb<br><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br></pre></td><td class="code"><pre><span class="line">subscriber.setProducer(<span class="keyword">new</span> Producer() &#123;</span><br><span class="line"></span><br><span class="line">            <span class="meta">@Override</span></span><br><span class="line">            <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">request</span><span class="params">(<span class="keyword">long</span> n)</span> </span>&#123;</span><br><span class="line">                AmbSubscriber&lt;T&gt; c;</span><br><span class="line">                <span class="keyword">if</span> ((c = selection.get()) != <span class="keyword">null</span>) &#123;</span><br><span class="line">                    <span class="comment">// propagate the request to that single Subscriber that won</span></span><br><span class="line">                    c.requestMore(n);</span><br><span class="line">                &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                    <span class="comment">//propagate the request to all the amb subscribers</span></span><br><span class="line">                    <span class="keyword">for</span> (AmbSubscriber&lt;T&gt; ambSubscriber: selection.ambSubscribers) &#123;</span><br><span class="line">                        <span class="keyword">if</span> (!ambSubscriber.isUnsubscribed()) &#123;</span><br><span class="line">                            <span class="comment">// make a best endeavours check to not waste requests</span></span><br><span class="line">                            <span class="comment">// if first emission has already occurred</span></span><br><span class="line">                            <span class="keyword">if</span> (selection.get() == ambSubscriber) &#123;</span><br><span class="line">                                ambSubscriber.requestMore(n);</span><br><span class="line">                                <span class="comment">// don't need to request from other subscribers because choice has been made</span></span><br><span class="line">                                <span class="comment">// and request has gone to choice</span></span><br><span class="line">                                <span class="keyword">return</span>;</span><br><span class="line">                            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                                ambSubscriber.requestMore(n);</span><br><span class="line">                            &#125;</span><br><span class="line">                        &#125;</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;);</span><br></pre></td></tr></table></figure></p>
<p>AmbSubscriber<br><figure class="highlight kotlin"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line"><span class="keyword">private</span> void requestMore(long n) &#123;</span><br><span class="line">            request(n);</span><br><span class="line">        &#125;</span><br><span class="line">...</span><br></pre></td></tr></table></figure></p>
<p>可以看到<code>AmbSubscriber</code>的<code>requestMore</code>就是去调用父类<code>Subscriber</code>的<code>request</code>方法。</p>
<p>OnSubscribe<br><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line">...</span><br><span class="line"> <span class="function"><span class="keyword">protected</span> <span class="keyword">final</span> <span class="keyword">void</span> <span class="title">request</span><span class="params">(<span class="keyword">long</span> n)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span> (n &lt; <span class="number">0</span>) &#123;</span><br><span class="line">            <span class="keyword">throw</span> <span class="keyword">new</span> IllegalArgumentException(<span class="string">"number requested cannot be negative: "</span> + n);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="comment">// if producer is set then we will request from it</span></span><br><span class="line">        <span class="comment">// otherwise we increase the requested count by n</span></span><br><span class="line">        Producer producerToRequestFrom;</span><br><span class="line">        <span class="keyword">synchronized</span> (<span class="keyword">this</span>) &#123;</span><br><span class="line">            <span class="keyword">if</span> (producer != <span class="keyword">null</span>) &#123;</span><br><span class="line">                producerToRequestFrom = producer;</span><br><span class="line">            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                addToRequested(n);</span><br><span class="line">                <span class="keyword">return</span>;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">// after releasing lock (we should not make requests holding a lock)</span></span><br><span class="line">        producerToRequestFrom.request(n);</span><br><span class="line">    &#125;</span><br><span class="line">...</span><br></pre></td></tr></table></figure></p>
<p><code>Subscriber</code>的<code>request</code>方法呢 其实主要就做1件事情，</p>
<ol>
<li>addToRequested<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">addToRequested</span><span class="params">(<span class="keyword">long</span> n)</span> </span>&#123;</span><br><span class="line">      <span class="keyword">if</span> (requested == NOT_SET) &#123;</span><br><span class="line">          requested = n;</span><br><span class="line">      &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">          <span class="keyword">final</span> <span class="keyword">long</span> total = requested + n;</span><br><span class="line">          <span class="comment">// check if overflow occurred</span></span><br><span class="line">          <span class="keyword">if</span> (total &lt; <span class="number">0</span>) &#123;</span><br><span class="line">              requested = Long.MAX_VALUE;</span><br><span class="line">          &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">              requested = total;</span><br><span class="line">          &#125;</span><br><span class="line">      &#125;</span><br><span class="line">  &#125;</span><br></pre></td></tr></table></figure>
</li>
</ol>
<p>其实就是记录了下总的请求数</p>
<ol start="2">
<li>producerToRequestFrom.request(n);</li>
</ol>
<p>由于我们现在demo很简单，只是单个操作符，所以如果跟入代码的话，可能就不会运行到这里，producerToRequestFrom大家可以看出来其实也是一个 <code>Producer</code>。这就我们就先不深入，先留个大概的印象，<code>RxJava</code>的源码有很多的这样的嵌套，<code>OnSubscribe</code>再包一层<code>OnSubscribe</code>,<code>Subscriber</code>再包一层<code>Subscriber</code>。</p>
<h6 id="总结"><a href="#总结" class="headerlink" title="总结"></a>总结</h6><p>其实总体下来，已经把<code>Amb</code>操作符的源码过了一遍，但是很可能大家还存在一些问题。比如我自己，我刚开始在看的时候就发现了一个问题。就是第五部分代码执行完毕后会怎么样呢？</p>
<p><code>Amb</code>操作符在前面介绍了，很可能是在异步的情况下操作，也就是说当<code>call</code>方法都执行完毕了，<code>AmbSubscriber</code>的<code>onNext</code>方法还是没有被执行，通过前面的代码我们可以看到，只有当第一个的<code>AmbSubscriber</code>的<code>onNext</code>方法执行后，其他<code>AmbSubscriber</code>才会被取消订阅。所以如果是异步操作的话，多个<code>AmbSubscriber</code>其实是一直存在于内存等待运行的。</p>
<p>所以在这里其实我们很容易看出，为什么我们最好要手动的去调用<code>unsubscribe</code>方法，因为在异步的情况下，是很容易造成内存泄漏的。</p>
<p>由于这篇文章有很多地方引用了上一篇文章的内容，如果没有看过上一篇文章的话，请先阅读下上一篇文章，这样能更好理解</p>
<p><a href="https://www.jianshu.com/p/e1d08ebd1571" target="_blank" rel="noopener">我是最简单的操作符—–Create</a></p>

      
    </div>
    
    
    

    

    
      <div>
        <div style="padding: 10px 0; margin: 20px auto; width: 90%; text-align: center;">
  <div>Donate comment here</div>
  <button id="rewardButton" disable="enable" onclick="var qr = document.getElementById('QR'); if (qr.style.display === 'none') {qr.style.display='block';} else {qr.style.display='none'}">
    <span>打赏</span>
  </button>
  <div id="QR" style="display: none;">

    
      <div id="wechat" style="display: inline-block">
        <img id="wechat_qr" src="/jexo/images/wechatpay.png" alt="Ling Hu 微信支付"/>
        <p>微信支付</p>
      </div>
    

    
      <div id="alipay" style="display: inline-block">
        <img id="alipay_qr" src="/jexo/images/alipay.png" alt="Ling Hu 支付宝"/>
        <p>支付宝</p>
      </div>
    

    

  </div>
</div>

      </div>
    

    

    <footer class="post-footer">
      
        <div class="post-tags">
          
            <a href="/jexo/tags/Android/" rel="tag"># Android</a>
          
            <a href="/jexo/tags/RxJava/" rel="tag"># RxJava</a>
          
            <a href="/jexo/tags/源码解析/" rel="tag"># 源码解析</a>
          
        </div>
      

      
      
      

      
        <div class="post-nav">
          <div class="post-nav-next post-nav-item">
            
              <a href="/jexo/2018/04/18/我是最简单的操作符-Create/" rel="next" title="我是最简单的操作符-----Create">
                <i class="fa fa-chevron-left"></i> 我是最简单的操作符-----Create
              </a>
            
          </div>

          <span class="post-nav-divider"></span>

          <div class="post-nav-prev post-nav-item">
            
              <a href="/jexo/2018/04/18/我是用来组合的操作符-Buffer/" rel="prev" title="我是用来组合的操作符------Buffer">
                我是用来组合的操作符------Buffer <i class="fa fa-chevron-right"></i>
              </a>
            
          </div>
        </div>
      

      
      
    </footer>
  </div>
  
  
  
  </article>



    <div class="post-spread">
      
    </div>
  </div>


          </div>
          


          

  
    <div class="comments" id="comments">
      
        <div id="gitment-container"></div>
      
    </div>

  



        </div>
        
          
  
  <div class="sidebar-toggle">
    <div class="sidebar-toggle-line-wrap">
      <span class="sidebar-toggle-line sidebar-toggle-line-first"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-last"></span>
    </div>
  </div>

  <aside id="sidebar" class="sidebar">
    
    <div class="sidebar-inner">

      

      
        <ul class="sidebar-nav motion-element">
          <li class="sidebar-nav-toc sidebar-nav-active" data-target="post-toc-wrap">
            文章目录
          </li>
          <li class="sidebar-nav-overview" data-target="site-overview-wrap">
            站点概览
          </li>
        </ul>
      

      <section class="site-overview-wrap sidebar-panel">
        <div class="site-overview">
          <div class="site-author motion-element" itemprop="author" itemscope itemtype="http://schema.org/Person">
            
              <img class="site-author-image" itemprop="image"
                src="/jexo/images/avatar.jpg"
                alt="Ling Hu" />
            
              <p class="site-author-name" itemprop="name">Ling Hu</p>
              <p class="site-description motion-element" itemprop="description">Android,源码解析,每天进步一点点</p>
          </div>

          <nav class="site-state motion-element">

            
              <div class="site-state-item site-state-posts">
              
                <a href="/jexo/archives/">
              
                  <span class="site-state-item-count">19</span>
                  <span class="site-state-item-name">日志</span>
                </a>
              </div>
            

            
              
              
              <div class="site-state-item site-state-categories">
                <a href="/jexo/categories/index.html">
                  <span class="site-state-item-count">9</span>
                  <span class="site-state-item-name">分类</span>
                </a>
              </div>
            

            
              
              
              <div class="site-state-item site-state-tags">
                <a href="/jexo/tags/index.html">
                  <span class="site-state-item-count">14</span>
                  <span class="site-state-item-name">标签</span>
                </a>
              </div>
            

          </nav>

          

          
            <div class="links-of-author motion-element">
                
                  <span class="links-of-author-item">
                    <a href="https://github.com/javalong" target="_blank" title="GitHub">
                      
                        <i class="fa fa-fw fa-github"></i>GitHub</a>
                  </span>
                
            </div>
          

          
          

          
          

          

        </div>
      </section>

      
      <!--noindex-->
        <section class="post-toc-wrap motion-element sidebar-panel sidebar-panel-active">
          <div class="post-toc">

            
              
            

            
              <div class="post-toc-content"><ol class="nav"><li class="nav-item nav-level-6"><a class="nav-link" href="#Amb操作符功能介绍"><span class="nav-number">1.</span> <span class="nav-text">Amb操作符功能介绍</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#用途"><span class="nav-number">2.</span> <span class="nav-text">用途</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#来一段小代码"><span class="nav-number">3.</span> <span class="nav-text">来一段小代码</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#看看源代码"><span class="nav-number">4.</span> <span class="nav-text">看看源代码</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#第二部分"><span class="nav-number">5.</span> <span class="nav-text">第二部分</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#第四部分"><span class="nav-number">6.</span> <span class="nav-text">第四部分</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#第五部分"><span class="nav-number">7.</span> <span class="nav-text">第五部分</span></a></li><li class="nav-item nav-level-6"><a class="nav-link" href="#总结"><span class="nav-number">8.</span> <span class="nav-text">总结</span></a></li></ol></div>
            

          </div>
        </section>
      <!--/noindex-->
      

      

    </div>
  </aside>


        
      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="footer-inner">
        <div class="copyright">&copy; <span itemprop="copyrightYear">2018</span>
  <span class="with-love">
    <i class="fa fa-user"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">Ling Hu</span>

  
</div>









        







        
      </div>
    </footer>

    
      <div class="back-to-top">
        <i class="fa fa-arrow-up"></i>
        
      </div>
    

    

  </div>

  

<script type="text/javascript">
  if (Object.prototype.toString.call(window.Promise) !== '[object Function]') {
    window.Promise = null;
  }
</script>









  












  
  
    <script type="text/javascript" src="/jexo/lib/jquery/index.js?v=2.1.3"></script>
  

  
  
    <script type="text/javascript" src="/jexo/lib/fastclick/lib/fastclick.min.js?v=1.0.6"></script>
  

  
  
    <script type="text/javascript" src="/jexo/lib/jquery_lazyload/jquery.lazyload.js?v=1.9.7"></script>
  

  
  
    <script type="text/javascript" src="/jexo/lib/velocity/velocity.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/jexo/lib/velocity/velocity.ui.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/jexo/lib/fancybox/source/jquery.fancybox.pack.js?v=2.1.5"></script>
  


  


  <script type="text/javascript" src="/jexo/js/src/utils.js?v=5.1.4"></script>

  <script type="text/javascript" src="/jexo/js/src/motion.js?v=5.1.4"></script>



  
  

  
  <script type="text/javascript" src="/jexo/js/src/scrollspy.js?v=5.1.4"></script>
<script type="text/javascript" src="/jexo/js/src/post-details.js?v=5.1.4"></script>



  


  <script type="text/javascript" src="/jexo/js/src/bootstrap.js?v=5.1.4"></script>



  


  




	





  





  







<!-- LOCAL: You can save these files to your site and update links -->
    
        
        <link rel="stylesheet" href="https://aimingoo.github.io/gitmint/style/default.css">
        <script src="https://aimingoo.github.io/gitmint/dist/gitmint.browser.js"></script>
    
<!-- END LOCAL -->

    
      <style>
        a.gitment-editor-footer-tip { display: none; }
        .gitment-container.gitment-footer-container { display: none; }
      </style>
    

    
      <script type="text/javascript">
      function renderGitment(){
        var gitment = new Gitmint({
            id: getId(),
            owner: 'javalong',
            repo: 'RxJava-Operator',
            
            lang: "zh-Hans" || navigator.language || navigator.systemLanguage || navigator.userLanguage,
            
            oauth: {
            
            
                client_secret: '06b042e57b8420f5002a6407348dde078bcde761',
            
                client_id: '1b5481d6512b2f28d7a2'
            }});
        gitment.render('gitment-container');
      }

      function getId(){
            if( window.location.pathname.length>=50){
                return window.location.pathname.substr(0,50)
            }else{
                return window.location.pathname
            }
      }

      
      renderGitment();
      
      </script>
    







  





  

  

  

  
  

  

  

  

</body>
</html>
